package com.uuuo.rxjava1;

import org.junit.Test;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

import rx.Notification;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.schedulers.TimeInterval;
import rx.schedulers.Timestamped;

/**
 * Created by JiaTao on 2017/4/5.
 */

public class F_Help extends A_Base {
    @Test
    public void materialize() {
        Observable.range(1, 10)
                .materialize().subscribe(new Subscriber<Notification<Integer>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Notification<Integer> integerNotification) {
                showMsg(integerNotification.getKind().name() + "----------" + integerNotification.getValue());
            }
        });
    }

    @Test
    public void timestamp() {
        Observable.range(1, 10)
                .timestamp()
                .subscribe(new Subscriber<Timestamped<Integer>>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Timestamped<Integer> integerTimestamped) {
                        String timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
                                .format(new Date(integerTimestamped.getTimestampMillis()));
                        showMsg(timestamp);
                    }
                });
    }

    @Test
    public void timeInterval() {
        Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 1; i < 10; i++) {
                    try {
                        Thread.sleep(100 * i);
                    } catch (InterruptedException e) {
                        subscriber.onError(e);
                    }
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        })
                .timeInterval()
//                .subscribeOn(Schedulers.newThread())
                .subscribe(new Subscriber<TimeInterval<Integer>>() {

                    @Override
                    public void onCompleted() {
                        System.out.println("onCompleted.");
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(TimeInterval<Integer> timeInterval) {
                        System.out.println("onNext: interval=" + timeInterval.getIntervalInMilliseconds() + ", Value=" + timeInterval.getValue());
                    }
                });
    }

    @Test
    public void timeOut() {
        Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 1; i < 4; i++) {
                    try {
                        Thread.sleep(100 * i);
                    } catch (InterruptedException e) {
                        subscriber.onError(e);
                    }
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        })
                .timeout(250, TimeUnit.MILLISECONDS)
//                .subscribeOn(Schedulers.newThread())
                .subscribe(new Subscriber<Integer>() {

                    @Override
                    public void onCompleted() {
                        System.out.println("onCompleted.");
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext: " + integer);
                    }
                });
    }

    @Test
    public void doOnEach() {
        Observable.range(1, 10)
                .doOnEach(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        showMsg("onNext" + integer);
                    }
                }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                showMsg("subscribe" + integer);
            }
        });
    }
}
